// Copyright 2012 Anno Langen. All Rights Reserved.

package org.anno;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.inject.AbstractModule;
import com.google.inject.Module;
import com.google.inject.Provides;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import javax.inject.Inject;
import javax.inject.Singleton;

/**
 * Ratings are stored in about 300,000 bytes of byte arrays, indexed by the product of the most
 * significant bits. Note that MAX_USER_ID < 41*2^16 and MAX_MOVIE_ID < 70*2^8.
 */
public class Ratings implements Serializable {

  static final long serialVersionUID = -347930126801369558L;
  public static final int MOVIE_SHARD_COUNT = 70;
  public static final int USER_SHARD_COUNT = 41;

  /**
   * Receives all scores for a specific user, followed by close(). All calls to a processor instance
   * come from a single thread.
   */
  interface UserProcessor {

    void close();

    void movieScore(int movie, int score);
  }

  /**
   * Receives all scores for a specific movie, followed by close(). All calls to a processor
   * instance come from a single thread.
   */
  interface MovieProcessor {

    void close();

    void userScore(int user, int score);
  }

  /**
   * Factory to furnish a processor for a given ID.
   *
   * @param <T> type of processor.
   */
  interface ProcessorFactory<T> {

    T get(int id);
  }

  private final byte[][] data;

  private Ratings(byte[][] data) {
    this.data = data;
  }

  /**
   * Returns module to configure a single Ratings instance based on downloaded training set, or from
   * condensed representation in Java Object Stream format.
   */
  public static Module getModule() {
    return new AbstractModule() {
      protected void configure() {
      }

      @Provides @Singleton
      Ratings ratings() throws IOException, InterruptedException, ClassNotFoundException {
        return new File("ratings.jos").exists() ? read("ratings.jos")
            : buildInitiallyAndSave("../download/training_set", "ratings.jos");
      }
    };
  }

  /**
   * Calls userScore(...) on the result of factory.get(movieId). Guarantees that ratings for a given
   * movie will be sent from a single thread. Specifically, processes all ratings in 70 shards, most
   * of which pertain to 256 movies. Each thread may keep up to 256 movie processors open.
   */
  public void sendByMovie(final ProcessorFactory<MovieProcessor> f) throws InterruptedException {
    invokeAll(getSendByMovieTasks(f));
  }

  /**
   * Calls userScore(...) on the result of factory.get(movieId). Guarantees that ratings for a given
   * movie will be sent from a single thread. Specifically, processes all ratings in 70 shards, most
   * of which pertain to 256 movies. Each thread may keep up to 256 movie processors open.
   */
  public List<Callable<Void>> getSendByMovieTasks(final ProcessorFactory<MovieProcessor> f) {
    List<Callable<Void>> tasks = new ArrayList<Callable<Void>>(MOVIE_SHARD_COUNT);
    for (int m = MOVIE_SHARD_COUNT; --m >= 0; ) {
      final int movieMsb = m << 8;
      final int start = m * 5;
      tasks.add(new Callable<Void>() {
        public Void call() throws Exception {
          MovieProcessor[] opened = new MovieProcessor[1 << 8];
          for (int u = USER_SHARD_COUNT; --u >= 0; ) {
            int userStart = u * MOVIE_SHARD_COUNT * 5 + start;
            int userMsb = u << 16;
            for (int s = 5; --s >= 0; ) {
              byte[] bytes = data[userStart + s];
              for (int j = bytes.length - 3; j >= 0; j -= 3) {
                int movieLsb = bytes[j] & 255;
                MovieProcessor p = opened[movieLsb];
                if (p == null) {
                  opened[movieLsb] = p = f.get(movieMsb + movieLsb);
                }
                p.userScore(userMsb + ((bytes[j + 1] & 255) << 8) + (bytes[j + 2] & 255), s + 1);
              }
            }
          }
          for (MovieProcessor processor : opened) {
            if (processor != null) {
              processor.close();
            }
          }
          return null;
        }
      });
    }
    return tasks;
  }

  /**
   * Calls movieScore(...) on the result of factory.get(userId). Guarantees that ratings for a given
   * user will be sent from a single thread. Specifically, processes all ratings in 41 shards, most
   * of which pertain to 65,536 users. Each thread may keep up to 65,536 user processors open.
   */
  public void sendByUser(final ProcessorFactory<UserProcessor> f) throws InterruptedException {
    invokeAll(getSendByUserTasks(f));
  }

  private static void invokeAll(List<Callable<Void>> tasks) throws InterruptedException {
    List<Future<?>> futures = Lists.newArrayList();
    for (Callable<Void> task : tasks) {
      futures.add(Parallel.EXECUTOR.submit(task));
    }
    for (Future<?> future : futures) {
      try {
        future.get();
      } catch (ExecutionException e) {
        throw new AssertionError(e);
      }
    }
  }

  /**
   * Calls movieScore(...) on the result of factory.get(userId). Guarantees that ratings for a given
   * user will be sent from a single thread. Specifically, processes all ratings in 41 shards, most
   * of which pertain to 65,536 users. Each thread may keep up to 65,536 user processors open.
   */
  public List<Callable<Void>> getSendByUserTasks(final ProcessorFactory<UserProcessor> f) {
    List<Callable<Void>> tasks = Lists.newArrayListWithCapacity(USER_SHARD_COUNT);
    for (int u = USER_SHARD_COUNT; --u >= 0; ) {
      final int userMsb = u << 16;
      final int start = u * MOVIE_SHARD_COUNT * 5;
      tasks.add(new Callable<Void>() {
        public Void call() {
          UserProcessor[] opened = new UserProcessor[1 << 16];
          for (int m = MOVIE_SHARD_COUNT; --m >= 0; ) {
            int movieMsb = m << 8;
            int movieStart = start + m * 5;
            for (int s = 5; --s >= 0; ) {
            byte[] bytes = data[movieStart + s];
            for (int j = bytes.length - 3; j >= 0; j -= 3) {
              int userLsb = ((bytes[j + 1] & 255) << 8) + (bytes[j + 2] & 255);
              UserProcessor processor = opened[userLsb];
              if (processor == null) {
                processor = opened[userLsb] = f.get(userMsb + userLsb);
              }
              processor.movieScore(movieMsb + (bytes[j] & 255), s + 1);
            }
          }
          }
          for (UserProcessor processor : opened) {
            if (processor != null) {
              processor.close();
            }
          }
          return null;
        }
      });
    }
    return tasks;
  }

  public static void main(String[] args) throws Exception {
    buildInitiallyAndSave("../download/training_set", "ratings.jos");
    Parallel.EXECUTOR.shutdown();
  }

  private static Ratings read(String objectStreamFile) throws IOException, ClassNotFoundException {
    ObjectInputStream ois = new ObjectInputStream(new FileInputStream(objectStreamFile));
    try {
      return (Ratings) ois.readObject();
    } finally {
      ois.close();
    }
  }

  private static Ratings buildInitiallyAndSave(String trainingSetDir, String objectStreamFile)
      throws IOException, InterruptedException {
    byte[][] data = new byte[5 * USER_SHARD_COUNT * MOVIE_SHARD_COUNT][];
    int initialSize = 3 * 3 * 100000000 / data.length + 1;
    for (int i = data.length; --i >= 0; ) {
      data[i] = new byte[initialSize];
    }
    final int[] size = new int[data.length];
    Parallel.EXECUTOR.invokeAll(getReadingTasks(new File(trainingSetDir).listFiles(), size, data));
    ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(objectStreamFile));
    Ratings ratings = new Ratings(trimToSize(data, size));
    oos.writeObject(ratings);
    oos.close();
    return ratings;
  }

  private static byte[][] trimToSize(byte[][] data, int[] size) {
    for (int i = data.length; --i >= 0; ) {
      if (size[i] < data[i].length) {
        data[i] = Arrays.copyOf(data[i], size[i]);
      }
    }
    return data;
  }

  private static List<Callable<Void>> getReadingTasks(final File[] trainingSet, final int[] size,
      final byte[][] data) {
    List<Callable<Void>> tasks = Lists.newArrayListWithCapacity(MOVIE_SHARD_COUNT);
    for (int i = MOVIE_SHARD_COUNT; --i >= 0; ) {
      final int movieMsb = i << 8;
      tasks.add(new Callable<Void>() {
        public Void call() throws Exception {
          for (int movieLsb = 256; --movieLsb >= 0; ) {
            int m = movieMsb + movieLsb;
            if (m < trainingSet.length) {
              try {
                readFile(size, trainingSet[m], data);
              } catch (FileNotFoundException e) {
                throw new AssertionError(e);
              }
            }
          }
          return null;
        }
      });
    }
    return tasks;
  }

  private static void readFile(int[] size, File file, byte[][] data) throws FileNotFoundException {
    Scanner scanner = new Scanner(file);
    scanner.useDelimiter("[^0-9]+");
    int movieId = scanner.nextInt();
    while (scanner.hasNextInt()) {
      int userId = scanner.nextInt();
      int score = scanner.nextInt();
      int year = scanner.nextInt();
      int month = scanner.nextInt();
      int day = scanner.nextInt();
      int shard = (userId >> 16) * 5 * MOVIE_SHARD_COUNT + (movieId >> 8) * 5 + score - 1;
      Preconditions.checkArgument(size[shard] < data[shard].length,
          "shard %s has more than %s ratings", shard, size[shard] / 3);
      int start = size[shard];
      byte[] bytes = data[shard];
      bytes[start] = (byte) movieId;
      bytes[start + 1] = (byte) (userId >> 8);
      bytes[start + 2] = (byte) userId;
      size[shard] = start + 3;
    }
    scanner.close();
  }
}
